package cn.com.sparkle.paxos.stablestorage;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.Map.Entry;
import java.util.PriorityQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.addprocess.AddRequestPackage;
import cn.com.sparkle.paxos.deamon.InstanceExecutor;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceVoteRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SuccessfulRecord;
import cn.com.sparkle.paxos.model.Value;
import cn.com.sparkle.paxos.stablestorage.ReadSuccessRecordCallback.SuccessfulRecordWrap;
import cn.com.sparkle.paxos.stablestorage.event.PrepareRecordRealWriteEvent;
import cn.com.sparkle.paxos.stablestorage.io.BufferedFileOut;
import cn.com.sparkle.paxos.stablestorage.model.Record;
import cn.com.sparkle.paxos.stablestorage.model.RecordBody;
import cn.com.sparkle.paxos.stablestorage.model.RecordHead;
import cn.com.sparkle.paxos.stablestorage.model.RecordType;

public class RecordFileOperator {
	private final static Logger logger = Logger
			.getLogger(RecordFileOperator.class);

	private final static int MAX_EXECUTED_SUCCESSFUL_RECORD_COUNT = 1000000;
	private final static int SPLIT_SUCCESSFUL_RECORD_COUNT = 1000;

	private InstanceExecutor instanceExecutor;
	private PriorityQueue<SuccessfulRecordWrap> unsafeRecordQueue = new PriorityQueue<SuccessfulRecordWrap>();
	private HashMap<Long, SuccessfulRecordWrap> unsafeSet = new HashMap<Long, SuccessfulRecordWrap>();
	private HashMap<Long, InstanceVoteRecord> votedInstanceRecordMap = new HashMap<Long, InstanceVoteRecord>();

	private HashSet<Long> badFileSet = new HashSet<Long>();
	private long startInstanceId = 0;
	private ReentrantLock writeLock = new ReentrantLock();
	private File dir;
	private LinkedList<RecordFileBean> recordFileBeanList = new LinkedList<RecordFileBean>();
	private volatile long lastExpectSafeInstanceId;
	private volatile long maxVoteInstanceId = -1;

	private static class RecordFileBean {
		private File file;
		private BufferedFileOut out;
		private long fileFlag;

		public RecordFileBean(File file) {
			super();
			this.file = file;
			fileFlag = Long.valueOf(file.getName());
		}
	}

	public RecordFileOperator(File dir, long lastExpectSafeInstanceId,
			InstanceExecutor instanceExecutor) throws ClassNotFoundException,
			IOException {
		this.dir = dir;
		this.lastExpectSafeInstanceId = lastExpectSafeInstanceId;
		this.maxVoteInstanceId = lastExpectSafeInstanceId - 1;
		this.instanceExecutor = instanceExecutor;
		if (!dir.exists()) {
			dir.mkdirs();
		}
	}

	/**
	 * 
	 * @return the last instanceId that can be executed
	 * @throws IOException
	 * @throws ClassNotFoundException
	 */
	public long init() throws IOException, ClassNotFoundException {
		File[] files = dir.listFiles();
		Comparator<File> comparator = new Comparator<File>() {
			@Override
			public int compare(File o1, File o2) {
				return Long.valueOf(o1.getName()).compareTo(
						Long.valueOf(o2.getName()));
			}
		};
		Arrays.sort(files, comparator);
		long successStartFileFlag = getFileFlagOfInstanceId(lastExpectSafeInstanceId);
		long pos = 0;
		// read from read log

		for (int i = 0; i < files.length; i++) {
			pos = 0;
			File f = files[i];
			RecordFileBean recordFileBean = new RecordFileBean(f);
			recordFileBeanList.addLast(recordFileBean);
			if (i == 0) {
				startInstanceId = recordFileBean.fileFlag
						* SPLIT_SUCCESSFUL_RECORD_COUNT;
			}
			long nextStartInstanceId = (recordFileBean.fileFlag + 1)
					* SPLIT_SUCCESSFUL_RECORD_COUNT;

			if (recordFileBean.fileFlag < successStartFileFlag) {
				continue;
			}
			DataInputStream in = new DataInputStream(new BufferedInputStream(
					new FileInputStream(f)));
			try {
				byte[] headBytes = new byte[20];
				byte[] checksumBytes = new byte[8];
				while (true) {
					try {
						int size = in.read(headBytes);
						if (size == headBytes.length) {
							RecordHead head = new RecordHead(headBytes);
							if (head.checkSum()) {
								int bodySize = head.getBodySize();
								if (head.getInstanceId() < lastExpectSafeInstanceId) {
									in.skipBytes(bodySize);
									pos += 20 + bodySize;
									continue;
								}

								byte[] bodyBytes = new byte[bodySize - 8];
								size = in.read(bodyBytes);
								if (size != bodyBytes.length) {
									break;
								}
								size = in.read(checksumBytes);
								if (size != checksumBytes.length) {
									break;
								}
								RecordBody body = new RecordBody(bodyBytes,
										checksumBytes);
								if (body.checkSum()) {
									if (head.getType() == RecordType.SUCCESS) {
										votedInstanceRecordMap.remove(head
												.getInstanceId());
										SuccessfulRecord record = SuccessfulRecord
												.newBuilder()
												.mergeFrom(body.getBody())
												.build();
										SuccessfulRecordWrap recordWrap = new SuccessfulRecordWrap(
												head.getInstanceId(), record,
												null);
										unsafeRecordQueue.add(recordWrap);
										unsafeSet.put(head.getInstanceId(),
												recordWrap);
										while ((recordWrap = unsafeRecordQueue
												.peek()) != null
												&& unsafeRecordQueue
												.peek().getInstanceId() == lastExpectSafeInstanceId) {
											++lastExpectSafeInstanceId;
											instanceExecutor
													.execute(recordWrap);
											unsafeRecordQueue.poll();
											unsafeSet.remove(recordWrap
													.getInstanceId());
										}
									} else {
										InstanceVoteRecord instance = InstanceVoteRecord
												.newBuilder()
												.mergeFrom(body.getBody())
												.build();
										votedInstanceRecordMap.put(
												head.getInstanceId(), instance);
										if (head.getInstanceId() > maxVoteInstanceId) {
											maxVoteInstanceId = head
													.getInstanceId();
										}
									}
								} else {
									badFileSet.add(recordFileBean.fileFlag);
									logger.warn("checksum error fileflag:" + recordFileBean.fileFlag + " file pos:" + pos);
									break;
								}

							} else {
								badFileSet.add(recordFileBean.fileFlag);
								logger.warn("checksum error fileflag:" + recordFileBean.fileFlag + " file pos:" + pos);
								break;
							}
							
							pos += 20 + head.getBodySize();
						} else {
							break;
						}
					} catch (EOFException e) {
						break;
					}
				}
				// check file count
				if (lastExpectSafeInstanceId != nextStartInstanceId) {
					RandomAccessFile raff = new RandomAccessFile(f, "rws");
					raff.seek(pos);
					recordFileBean.out = new BufferedFileOut(raff);
				}
			} finally {
				close(in);
			}
		}
		if (recordFileBeanList.size() == 0) {
			File f = new File(dir + "/0");
			f.createNewFile();
			recordFileBeanList.add(new RecordFileBean(f));
		}
		return lastExpectSafeInstanceId - 1;
	}

	public boolean writeSuccessfulRecord(long instanceId,
			SuccessfulRecord successfulRecord,
			LinkedList<AddRequestPackage> addRequestPackages,
			final Callable<Object> realEvent) throws IOException {

		SuccessfulRecordWrap recordWrap = new SuccessfulRecordWrap(instanceId,
				successfulRecord, addRequestPackages);
		try {
			writeLock.lock();
			if (instanceId < lastExpectSafeInstanceId
					|| unsafeSet.containsKey(instanceId)) {
				try {
					realEvent.call();
				} catch (Exception e) {
				}
				return true;
			}
			votedInstanceRecordMap.remove(instanceId);
			unsafeSet.put(recordWrap.getInstanceId(), recordWrap);
			unsafeRecordQueue.add(recordWrap);
			while ((recordWrap = unsafeRecordQueue.peek()) != null
					&& recordWrap.getInstanceId() == lastExpectSafeInstanceId) {
				instanceExecutor.execute(recordWrap);
				++lastExpectSafeInstanceId;
				unsafeRecordQueue.poll();
				unsafeSet.remove(recordWrap.getInstanceId());
			}

			RecordBody body = new RecordBody(successfulRecord.toByteArray());
			RecordHead head = new RecordHead(body.getBody().length + 8,
					instanceId, RecordType.SUCCESS);
			Record record = new Record(head, body);
			return writeRecordLog(record, new Callable<Object>() {
				@Override
				public Object call() throws Exception {
					// notify
					if (realEvent != null) {
						realEvent.call();
					}
					return null;
				}
			});

		} finally {
			writeLock.unlock();
		}
	}

	/**
	 * 
	 * @param instanceId
	 * @param highestJoinNum
	 * @param highestVotedNum
	 * @param highestValue
	 * @param realWriteEvent
	 * @return is can write
	 * @throws IOException
	 */
	public boolean writeVoteRecord(final long instanceId,
			final InstanceVoteRecord record,
			final PrepareRecordRealWriteEvent realWriteEvent)
			throws IOException {
		try {
			writeLock.lock();
			if (instanceId < lastExpectSafeInstanceId
					|| unsafeSet.containsKey(instanceId)) {
				try {
					realWriteEvent.call(instanceId, null);
				} catch (Exception e) {
				}
				return true;
			}

			if (instanceId > maxVoteInstanceId) {
				maxVoteInstanceId = instanceId;
			}
			// record vote log to map
			votedInstanceRecordMap.put(instanceId, record);

			RecordBody body = new RecordBody(record.toByteArray());
			RecordHead head = new RecordHead(body.getBody().length + 8,
					instanceId, RecordType.VOTE);
			Record r = new Record(head, body);

			return writeRecordLog(r, new Callable<Object>() {
				@Override
				public Object call() throws Exception {
					if (realWriteEvent != null) {
						realWriteEvent.call(instanceId, record);
					}
					return null;
				}
			});
		} finally {
			writeLock.unlock();
		}
	}

	private boolean writeRecordLog(Record record, Callable<Object> callable)
			throws FileNotFoundException, IOException {
		RecordFileBean curFb = null;
		BufferedFileOut out = null;
		long fileFlag = getFileFlagOfInstanceId(record.getHead()
				.getInstanceId());
		if(badFileSet.contains(fileFlag) && record.getHead().getType() != RecordType.SUCCESS){//successful record must be write
			return false;
		}
		long nextInstanceId = (fileFlag + 1) * SPLIT_SUCCESSFUL_RECORD_COUNT;
		if (recordFileBeanList.getLast().fileFlag < fileFlag) {
			for (long i = recordFileBeanList.getLast().fileFlag + 1; i <= fileFlag; ++i) {
				File f = new File(dir + "/" + i);
				f.createNewFile();
				RecordFileBean recordFileBean = new RecordFileBean(f);
				recordFileBeanList.addLast(recordFileBean);
			}
		}

		Iterator<RecordFileBean> iter = recordFileBeanList.descendingIterator();
		while (iter.hasNext()) {
			curFb = iter.next();
			if (curFb.fileFlag == fileFlag) {
				out = curFb.out;
				if (out == null) {
					curFb.out = new BufferedFileOut(new RandomAccessFile(
							curFb.file, "rws"));
					out = curFb.out;
				}
				break;
			}
		}
		record.writeToStream(out, callable);

		// close full file
		if (lastExpectSafeInstanceId >= nextInstanceId) {
			badFileSet.remove(fileFlag);
			if (curFb.out != null) {
				curFb.out.close();
				curFb.out = null;
			}
		}
		return true;
	}

	public long getLastExpectSafeInstanceId() {
		return lastExpectSafeInstanceId;
	}

	public HashMap<Long, InstanceVoteRecord> getVotedInstanceRecordMap() {
		return votedInstanceRecordMap;
	}

	public long getMaxVoteInstanceId() {
		return maxVoteInstanceId;
	}

	public void finishInstance(long instanceId) {
		try {
			writeLock.lock();
			long executedCount = instanceId - startInstanceId + 1;
			// delete successful
			while (executedCount > MAX_EXECUTED_SUCCESSFUL_RECORD_COUNT) {
				long nextInstanceId = (recordFileBeanList.getFirst().fileFlag + 1)
						* SPLIT_SUCCESSFUL_RECORD_COUNT;
				if (lastExpectSafeInstanceId < nextInstanceId) {
					break;
				}
				executedCount -= SPLIT_SUCCESSFUL_RECORD_COUNT;
				RecordFileBean curFb = recordFileBeanList.removeFirst();
				if (!curFb.file.delete()) {
					break;
				}
				executedCount = nextInstanceId;
			}
		} finally {
			writeLock.unlock();
		}
	}

	public boolean isSuccessful(long instanceId) {
		if (instanceId < lastExpectSafeInstanceId) {
			return true;
		}
		try {
			writeLock.lock();
			if (instanceId < lastExpectSafeInstanceId
					|| unsafeSet.containsKey(instanceId)) {
				return true;
			} else {
				return false;
			}
		} finally {
			writeLock.unlock();
		}
	}

	public void read(long fromInstanceId, long toInstanceId,
			ReadSuccessRecordCallback successCallback) throws IOException {
			File[] files = dir.listFiles();
			Comparator<File> comparator = new Comparator<File>() {
				@Override
				public int compare(File o1, File o2) {
					return Long.valueOf(o1.getName()).compareTo(
							Long.valueOf(o2.getName()));
				}
			};
			Arrays.sort(files, comparator);
			long startFile = getFileFlagOfInstanceId(fromInstanceId);
			long endFile = getFileFlagOfInstanceId(toInstanceId);
			// read from read log

			for (int i = 0; i < files.length; i++) {
				long pos = 0;
				File f = files[i];
				RecordFileBean successfulFileBean = new RecordFileBean(f);
				if (successfulFileBean.fileFlag < startFile) {
					continue;
				} else if (successfulFileBean.fileFlag > endFile) {
					break;
				}

				DataInputStream in = new DataInputStream(
						new BufferedInputStream(new FileInputStream(f)));
				try {
					byte[] headBytes = new byte[20];
					byte[] checksumBytes = new byte[8];
					while (true) {
						try {
							int size = in.read(headBytes);
							if (size == headBytes.length) {
								RecordHead head = new RecordHead(headBytes);
								if (head.checkSum()) {
									int bodySize = head.getBodySize();
									if (head.getInstanceId() < fromInstanceId
											|| head.getInstanceId() > toInstanceId) {
										in.skipBytes(bodySize);
										pos += 20 + bodySize;
										continue;
									}

									byte[] bodyBytes = new byte[bodySize - 8];
									size = in.read(bodyBytes);
									if (size != bodyBytes.length) {
										break;
									}
									size = in.read(checksumBytes);
									if (size != checksumBytes.length) {
										break;
									}
									RecordBody body = new RecordBody(bodyBytes,
											checksumBytes);
									if (body.checkSum()) {
										if (head.getType() == RecordType.SUCCESS) {
											SuccessfulRecord record = SuccessfulRecord
													.newBuilder()
													.mergeFrom(body.getBody())
													.build();
											SuccessfulRecordWrap recordWrap = new SuccessfulRecordWrap(
													head.getInstanceId(),
													record, null);
											successCallback.read(recordWrap);
										}
										pos += 20 + bodySize;
									} else {
										logger.warn("checksum error fileflag:" + successfulFileBean.fileFlag + " file pos:" + pos);
										break;
									}

								} else {
									logger.warn("checksum error fileflag:" + successfulFileBean.fileFlag + " file pos:" + pos);
									break;

								}
							}else{
								break;
							}
						} catch (EOFException e) {
							break;
						}
					}
				} finally {
					close(in);
				}
			}
	}

	public long getFirstInstanceIdInUnsafe() {
		try {
			writeLock.lock();
			SuccessfulRecordWrap srw = unsafeRecordQueue.peek();
			return srw == null ? -1 : srw.getInstanceId();
		} finally {
			writeLock.unlock();
		}
	}

	private long getFileFlagOfInstanceId(long instanceId) {
		return instanceId / SPLIT_SUCCESSFUL_RECORD_COUNT;
	}

	private void close(InputStream is) {
		if (is != null) {
			try {
				is.close();
			} catch (IOException e) {
			}
		}
	}

}
